package com.fanshuai.spring.boot.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Consumer;

@Service
@Slf4j
public class CacheService {
    @Resource
    private RedisTemplate<String, String> redisTemplate;
    @Resource
    private RedisMessageListenerContainer listenerContainer;

    public boolean set(String key, String value) {
        redisTemplate.opsForValue().set(key, value);
        return true;
    }

    public String get(String key) {
        return redisTemplate.opsForValue().get(key);
    }

    public void zadd(String key, String value, double score) {
        redisTemplate.opsForZSet().add(key, value, score);
    }

    public Set<String> zrange(String key, long start, long end) {
        return redisTemplate.opsForZSet().range(key, start, end);
    }

    /**
     * pub/sub
     * @param channel
     * @param msg
     * @return
     */
    public boolean publish(String channel, String msg) {
        redisTemplate.convertAndSend(channel, msg);
        return true;
    }

    public void setPipelined(Map<String, String> data) {
        StringRedisSerializer serializer = new StringRedisSerializer();
        redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                for (Map.Entry<String, String> entry : data.entrySet()) {
                    byte[] key = serializer.serialize(entry.getKey());
                    byte[] value = serializer.serialize(entry.getValue());
                    redisConnection.stringCommands().set(key, value);
                }

                return null;
            }
        });
    }

    public <T> T executeLua(String script, List<String> keys, List<Object> args, Class<T> returnType) {
        DefaultRedisScript<T> redisScript = (DefaultRedisScript<T>) RedisScript.of(script);
        redisScript.setResultType(returnType);

        return redisTemplate.execute(redisScript, keys, args.toArray());
    }
}
